// Copyright 2021-present The Atlas Authors. All rights reserved.
// This source code is licensed under the Apache 2.0 license found
// in the LICENSE file in the root directory of this source tree.

package dm

import (
	"context"
	"database/sql"
	"fmt"
	"net/url"
	"strings"
	"time"

	"gitee.com/damengde/atlas/sql/internal/sqlx"
	"gitee.com/damengde/atlas/sql/migrate"

	"gitee.com/damengde/atlas/sql/schema"
	"gitee.com/damengde/atlas/sql/sqlclient"
)

type (
	// Driver represents a MySQL driver for introspecting database schemas,
	// generating diff between schema elements and apply migrations changes.
	Driver struct {
		*conn
		schema.Differ
		schema.Inspector
		migrate.PlanApplier
	}

	// database connection and its information.
	conn struct {
		schema.ExecQuerier
		// The schema was set in the path (schema connection).
		schema string
		// System variables that are set on `Open`.
		collate string
		charset string
		lcnames int
	}
)

var _ interface {
	migrate.Snapshoter
	migrate.StmtScanner
	migrate.CleanChecker
	schema.TypeParseFormatter
} = (*Driver)(nil)

// DriverName holds the name used for registration.
const DriverName = "dm"

func init() {
	sqlclient.Register(
		DriverName,
		sqlclient.OpenerFunc(opener),
		sqlclient.RegisterDriverOpener(Open),
		sqlclient.RegisterCodec(MarshalHCL, EvalHCL),
		sqlclient.RegisterFlavours("dameng", "Dameng"),
		sqlclient.RegisterURLParser(parser{}),
	)
}

// Open opens a new MySQL driver.
func Open(db schema.ExecQuerier) (migrate.Driver, error) {
	c := &conn{ExecQuerier: db}
	return &Driver{
		conn:        c,
		Differ:      &sqlx.Diff{DiffDriver: &diff{conn: c}},
		Inspector:   &inspect{c},
		PlanApplier: &planApply{c},
	}, nil
}

func opener(_ context.Context, u *url.URL) (*sqlclient.Client, error) {
	ur := parser{}.ParseURL(u)
	db, err := sql.Open(DriverName, ur.DSN)
	if err != nil {
		return nil, err
	}
	drv, err := Open(db)
	if err != nil {
		if cerr := db.Close(); cerr != nil {
			err = fmt.Errorf("%w: %v", err, cerr)
		}
		return nil, err
	}
	drv.(*Driver).schema = ur.Schema
	return &sqlclient.Client{
		Name:   DriverName,
		DB:     db,
		URL:    ur,
		Driver: drv,
	}, nil
}

// NormalizeRealm returns the normal representation of the given database.
func (d *Driver) NormalizeRealm(ctx context.Context, r *schema.Realm) (*schema.Realm, error) {
	return (&sqlx.DevDriver{Driver: d}).NormalizeRealm(ctx, r)
}

// NormalizeSchema returns the normal representation of the given database.
func (d *Driver) NormalizeSchema(ctx context.Context, s *schema.Schema) (*schema.Schema, error) {
	return (&sqlx.DevDriver{Driver: d}).NormalizeSchema(ctx, s)
}

// Lock implements the schema.Locker interface.
func (d *Driver) Lock(ctx context.Context, name string, timeout time.Duration) (schema.UnlockFunc, error) {
	conn, err := sqlx.SingleConn(ctx, d.ExecQuerier)
	if err != nil {
		return nil, err
	}
	if err := acquire(ctx, conn, name, timeout); err != nil {
		conn.Close()
		return nil, err
	}
	return func() error {
		defer conn.Close()
		rows, err := conn.QueryContext(ctx, "SELECT RELEASE_LOCK(?)", name)
		if err != nil {
			return err
		}
		switch released, err := sqlx.ScanNullBool(rows); {
		case err != nil:
			return err
		case !released.Valid || !released.Bool:
			return fmt.Errorf("sql/mysql: failed releasing a named lock %q", name)
		}
		return nil
	}, nil
}

// Snapshot implements migrate.Snapshoter.
func (d *Driver) Snapshot(ctx context.Context) (migrate.RestoreFunc, error) {
	// If the connection is bound to a schema, we can restore the state if the schema has no tables.
	s, err := d.InspectSchema(ctx, "", nil)
	if err != nil && !schema.IsNotExistError(err) {
		return nil, err
	}
	// If a schema was found, it has to have no tables attached to be considered clean.
	if s != nil {
		if len(s.Tables) > 0 {
			return nil, &migrate.NotCleanError{
				State:  schema.NewRealm(s),
				Reason: fmt.Sprintf("found table %q in schema %q", s.Tables[0].Name, s.Name),
			}
		}
		return d.SchemaRestoreFunc(s), nil
	}
	// Otherwise, the database can not have any schema.
	realm, err := d.InspectRealm(ctx, nil)
	if err != nil {
		return nil, err
	}
	if len(realm.Schemas) > 0 {
		return nil, &migrate.NotCleanError{State: realm, Reason: fmt.Sprintf("found schema %q", realm.Schemas[0].Name)}
	}
	return d.RealmRestoreFunc(realm), nil
}

// SchemaRestoreFunc returns a function that restores the given schema to its desired state.
func (d *Driver) SchemaRestoreFunc(desired *schema.Schema) migrate.RestoreFunc {
	return func(ctx context.Context) error {
		current, err := d.InspectSchema(ctx, desired.Name, nil)
		if err != nil {
			return err
		}
		changes, err := d.SchemaDiff(current, desired)
		if err != nil {
			return err
		}
		return d.ApplyChanges(ctx, changes)
	}
}

// RealmRestoreFunc returns a function that restores the given realm to its desired state.
func (d *Driver) RealmRestoreFunc(desired *schema.Realm) migrate.RestoreFunc {
	return func(ctx context.Context) error {
		current, err := d.InspectRealm(ctx, nil)
		if err != nil {
			return err
		}
		changes, err := d.RealmDiff(current, desired)
		if err != nil {
			return err
		}
		return d.ApplyChanges(ctx, changes)
	}
}

// CheckClean implements migrate.CleanChecker.
func (d *Driver) CheckClean(ctx context.Context, revT *migrate.TableIdent) error {
	if revT == nil { // accept nil values
		revT = &migrate.TableIdent{}
	}
	s, err := d.InspectSchema(ctx, "", nil)
	if err != nil && !schema.IsNotExistError(err) {
		return err
	}
	if s != nil {
		if len(s.Tables) == 0 || (revT.Schema == "" || s.Name == revT.Schema) && len(s.Tables) == 1 && s.Tables[0].Name == revT.Name {
			return nil
		}
		return &migrate.NotCleanError{
			State:  schema.NewRealm(s),
			Reason: fmt.Sprintf("found table %q in schema %q", s.Tables[0].Name, s.Name),
		}
	}
	r, err := d.InspectRealm(ctx, nil)
	if err != nil {
		return err
	}
	switch n := len(r.Schemas); {
	case n > 1:
		return &migrate.NotCleanError{State: r, Reason: fmt.Sprintf("found multiple schemas: %d", len(r.Schemas))}
	case n == 1 && r.Schemas[0].Name != revT.Schema:
		return &migrate.NotCleanError{State: r, Reason: fmt.Sprintf("found schema %q", r.Schemas[0].Name)}
	case n == 1 && len(r.Schemas[0].Tables) > 1:
		return &migrate.NotCleanError{State: r, Reason: fmt.Sprintf("found multiple tables: %d", len(r.Schemas[0].Tables))}
	case n == 1 && len(r.Schemas[0].Tables) == 1 && r.Schemas[0].Tables[0].Name != revT.Name:
		return &migrate.NotCleanError{State: r, Reason: fmt.Sprintf("found table %q", r.Schemas[0].Tables[0].Name)}
	}
	return nil
}

// Version returns the version of the connected database.
func (d *Driver) Version() string {
	return "dm v8"
}

// FormatType converts schema type to its column form in the database.
func (*Driver) FormatType(t schema.Type) (string, error) {
	return FormatType(t)
}

// ParseType returns the schema.Type value represented by the given string.
func (*Driver) ParseType(s string) (schema.Type, error) {
	return Parse(s)
}

// StmtBuilder is a helper method used to build statements with MySQL formatting.
func (*Driver) StmtBuilder(opts migrate.PlanOptions) *sqlx.Builder {
	return &sqlx.Builder{
		QuoteOpening: '`',
		QuoteClosing: '`',
		Schema:       opts.SchemaQualifier,
		Indent:       opts.Indent,
	}
}

// ScanStmts implements migrate.StmtScanner.
func (*Driver) ScanStmts(input string) ([]*migrate.Stmt, error) {
	return (&migrate.Scanner{
		ScannerOptions: migrate.ScannerOptions{
			MatchBegin: true,
			// The following are not support by MySQL/MariaDB.
			MatchBeginAtomic: false,
			MatchDollarQuote: false,
		},
	}).Scan(input)
}

func acquire(ctx context.Context, conn schema.ExecQuerier, name string, timeout time.Duration) error {
	rows, err := conn.QueryContext(ctx, "SELECT GET_LOCK(?, ?)", name, int(timeout.Seconds()))
	if err != nil {
		return err
	}
	switch acquired, err := sqlx.ScanNullBool(rows); {
	case err != nil:
		return err
	case !acquired.Valid:
		// NULL is returned in case of an unexpected internal error.
		return fmt.Errorf("sql/mysql: unexpected internal error on Lock(%q, %s)", name, timeout)
	case !acquired.Bool:
		return schema.ErrLocked
	}
	return nil
}

// unescape strings with backslashes returned
// for SQL expressions from information schema.
func unescape(s string) string {
	var b strings.Builder
	for i := 0; i < len(s); i++ {
		switch c := s[i]; {
		case c != '\\' || i == len(s)-1:
			b.WriteByte(c)
		case s[i+1] == '\'', s[i+1] == '\\':
			b.WriteByte(s[i+1])
			i++
		}
	}
	return b.String()
}

type parser struct{}

// ParseURL implements the sqlclient.URLParser interface.
func (parser) ParseURL(u *url.URL) *sqlclient.URL {
	v := u.Query()
	v.Set("parseTime", "true")
	u.RawQuery = v.Encode()
	return &sqlclient.URL{URL: u, DSN: dsn(u), Schema: strings.TrimPrefix(u.Path, "/")}
}

// ChangeSchema implements the sqlclient.SchemaChanger interface.
func (parser) ChangeSchema(u *url.URL, s string) *url.URL {
	nu := *u
	nu.Path = "/" + s
	return &nu
}

// dsn returns the MySQL standard DSN for opening
// the sql.DB from the user provided URL.
func dsn(u *url.URL) string {
	var (
		b      strings.Builder
		values = u.Query()
	)
	b.WriteString(u.User.Username())
	if p, ok := u.User.Password(); ok {
		b.WriteByte(':')
		b.WriteString(p)
	}
	if b.Len() > 0 {
		b.WriteByte('@')
	}
	switch {
	case strings.HasSuffix(u.Scheme, "+unix"):
		b.WriteString("unix(")
		// The path is always absolute, and
		// therefore the host should be empty.
		b.WriteString(u.Path)
		b.WriteString(")/")
		if name := values.Get("database"); name != "" {
			b.WriteString(name)
			values.Del("database")
		}
	default:
		if u.Host != "" {
			b.WriteString("tcp(")
			b.WriteString(u.Host)
			b.WriteByte(')')
		}
		if u.Path != "" {
			b.WriteString(u.Path)
		} else {
			b.WriteByte('/')
		}
	}
	if p := values.Encode(); p != "" {
		b.WriteByte('?')
		b.WriteString(p)
	}
	return b.String()
}

const (
	// 常规数据类型
	TypeChar            = "CHAR"             // 定长字符串，最大长度由数据库页面大小决定。
	TypeCharacter       = "CHARACTER"        // 定长字符串，最大长度由数据库页面大小决定。
	TypeVarchar         = "VARCHAR"          // 可变长字符串，最大长度由数据库页面大小决定。
	TypeNumeric         = "NUMERIC"          // 用于存储零、正负定点数。精度范围1至38
	TypeNumber          = "NUMBER"           // 跟NUMERIC相同
	TypeDecimal         = "DECIMAL"          // 跟NUMERIC相似
	TypeDec             = "DEC"              // 跟NUMERIC相似
	TypeInteger         = "INTEGER"          // 用于存储有符号整数，精度为10
	TypeInt             = "INT"              // 用于存储有符号整数，精度为10
	TypeBigint          = "BIGINT"           // 用于存储有符号整数，精度为19，标度为0
	TypeTinyint         = "TINYINT"          // 用于存储有符号整数，精度为3，标度为0。取值范围为：-128~+127
	TypeByte            = "BYTE"             // 与TINYINT相似，精度为3，标度为0
	TypeSmallint        = "SMALLINT"         // 用于存储有符号整数，精度为5，标度为0
	TypeBinary          = "BINARY"           // 指定定长二进制数据
	TypeVarbinary       = "VARBINARY"        // 指定变长二进制数据。用法类似BINARY数据类型
	TypeReal            = "REAL"             // 带二进制的浮点数
	TypeFloat           = "FLOAT"            // 二进制精度的浮点数，精度最大不超过53
	TypeDouble          = "DOUBLE"           // 同FLOAT相似，精度最大不超过53
	TypeDoublePrecision = "DOUBLE PRECISION" // 该类型指明双精度浮点数，其二进制精度为53，十进制精度为15

	//位串数据类型
	TypeBit = "BIT" // 用于存储整数数据1、0或null

	//日期时间数据类型
	TypeDate                       = "DATE"                           // 包括年、月、日信息，定义了‘-4712-01-01’和‘9999-12-31’之间任何一个有效的格里高利日期
	TypeTime                       = "TIME"                           // 包括时、分、秒信息
	TypeTimestamp                  = "TIMESTAMP"                      // 包括年、月、日、时、分、秒信息
	TypeDatetime                   = "DATETIME"                       // 包括年、月、日、时、分、秒信息
	TypeTimeWithTimeZone           = "TIME WITH TIME ZONE"            // 描述一个带时区的TIME值
	TypeTimestampWithTimeZone      = "TIMESTAMP WITH TIME ZONE"       // 描述一个带时区的TIMESTAMP值
	TypeTimestampWithLocalTimeZone = "TIMESTAMP WITH LOCAL TIME ZONE" // 描述一个本地时区的TIMESTAMP值

	//多媒体数据类型
	TypeText          = "TEXT"          // 变长字符串类型，其字符串的长度最大为2G-1，可用于存储长的文本串
	TypeLongvarchar   = "LONGVARCHAR"   // 变长字符串类型，其字符串的长度最大为2G-1，可用于存储长的文本串
	TypeImage         = "IMAGE"         // 可用于存储多媒体信息中的图像类型
	TypeLongvarbinary = "LONGVARBINARY" // 可用于存储多媒体信息中的图像类型
	TypeBlob          = "BLOB"          // 用于指明变长的字符串，长度最大为2G-1字节
	TypeClob          = "CLOB"          // 用于指明变长的字符串，长度最大为2G-1字节
	TypeBfile         = "BFILE"         // 用于指明存储在操作系统中的二进制文件

)

// Additional common constants in MySQL.
const (
	IndexTypeBTree    = "BTREE"
	IndexTypeHash     = "HASH"
	IndexTypeFullText = "FULLTEXT"
	IndexTypeSpatial  = "SPATIAL"

	IndexParserNGram = "ngram"
	IndexParserMeCab = "mecab"

	EngineInnoDB = "InnoDB"
	EngineMyISAM = "MyISAM"
	EngineMemory = "Memory"
	EngineCSV    = "CSV"
	EngineNDB    = "NDB" // NDBCLUSTER

	currentTS     = "current_timestamp"
	defaultGen    = "default_generated"
	autoIncrement = "auto_increment"

	virtual    = "VIRTUAL"
	stored     = "STORED"
	persistent = "PERSISTENT"
)
